package com.wuba.rxjavademo.lyz.day0416;

import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.schedulers.Schedulers;

/**
 * Created by liyaxing on 2016/4/18.
 */
public class Demo04 {


    public static void main(String[] ss){

        Demo04 demo04 = new Demo04() ;
//        Observable<Integer> observable =
//        Observable.combineLatest(demo04.createObservable(1),demo04.createObservable(1),(num1,num2)-> num1 + num2) ;
//        observable.subscribe(integer -> print(integer)) ;
        Observable<Integer> observable = demo04.errorReturnObservable() ;
        observable.subscribe(i -> print(i)) ;

        try {
            Thread.sleep(50000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private Observable<Integer> errorReturnObservable(){
        return createObservable(1).onErrorReturn(throwable -> 999) ;
    }

    private Observable<String> zipObservable(){
        return Observable.zip(Observable.just(1,2,3),Observable.just(4,5,6),(n1,n2) -> n1 + "-" + n2) ;
    }

    private Observable<String> zipWithObservable(){
        return Observable.just(1,2,3,4).zipWith(Observable.just(4,5,6),(n1,n2) -> n1 + "-" + n2) ;
    }

    private Observable<Integer> switchObservable(){
        return Observable.switchOnNext(Observable.create(subscriber -> {

            for (int i = 0; i < 5; i++) {
                subscriber.onNext(createObservable(i));
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            subscriber.onCompleted();

        })) ;
    }

    private Observable<Integer> startWithObservable(){
        return Observable.just(1,2,3,4).startWith(-2,-1) ;
    }

    private Observable<Integer> mergeDelayErrorObservable(){
        return Observable.mergeDelayError(Observable.create(subscriber -> {
            for (int i = 0; i < 20; i++) {
                if(i == 3){
                    subscriber.onError(new Exception("value = 3"));
                }else {
                    subscriber.onNext(i);
                }
            }
            subscriber.onCompleted();
        }),Observable.just(1,2,3,4,5)) ;
    }

    private Observable<Integer> mergeObservable(){
        return Observable.just(1,2,3).mergeWith(Observable.just(4,5,6)) ;
    }


    private Observable<Observable<String>> groupJoinObservable(){
        return Observable.just(1,2,3).groupJoin(createObservable(1),
                integer -> Observable.timer(3000, TimeUnit.MILLISECONDS),
                integer -> Observable.timer(300, TimeUnit.MILLISECONDS),
                (num1,observable) -> observable.map(integer1 -> "left : " + num1 + " right : " + integer1))  ;
    }

    private Observable<String> joinObservable(){
        return Observable.just(1,2,3).join(createObservable(1),
                integer -> Observable.timer(3000, TimeUnit.MILLISECONDS),
                integer -> Observable.timer(300, TimeUnit.MILLISECONDS),
                (num1,num2) -> "left : " + num1 + " right : " + num2) ;
    }

    private Observable<Integer> createObservable(int index){

        return Observable.create(subscriber -> {
            for (int i = 0; i < 20; i++) {
                subscriber.onNext(i * index);

                if(i == 3){
                    subscriber.onError(new Throwable("value = 3"));
                }

//                try {
//                    Thread.sleep(200);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
            }
            subscriber.onCompleted();
        }).subscribeOn(Schedulers.newThread()).cast(Integer.class) ;

    }

    private static void print(Object obj){
        System.out.println(obj) ;
    }

}
